package cas.ihep.test;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class WordCount extends Configured implements Tool {

    static class WCMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
        final IntWritable one=new IntWritable(1);
        protected void map(LongWritable lno,Text line,Context context) throws IOException, InterruptedException {
            //拿到一行文本内容，转换成String 类型
            String linestr = line.toString();
            //将这行文本切分成单词
            String[] words=linestr.split(" ");
            //输出<单词，1>
            for(String word:words){
                context.write(new Text(word), one);
            }
        }
    }

    static class WCReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
        protected void reduce(Text key,Iterable<IntWritable> values,Reducer.Context context) throws IOException, InterruptedException {
            //定义一个计数器
            int count = 0;
            //通过value这个迭代器，遍历这一组kv中所有的value，进行累加
            for(IntWritable value:values){
                count+=value.get();
            }
            //输出这个单词的统计结果
            context.write(key, new IntWritable(count));
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Job wordCountJob = Job.getInstance(getConf());
        ProcessBuilder builder=new ProcessBuilder();
        //重要：指定本job所在的jar包
        wordCountJob.setJarByClass(WordCount.class);

        //设置wordCountJob所用的mapper逻辑类为哪个类
        wordCountJob.setMapperClass(WCMapper.class);
        //设置wordCountJob所用的reducer逻辑类为哪个类
        wordCountJob.setReducerClass(WCReducer.class);

        //设置map阶段输出的kv数据类型
        wordCountJob.setMapOutputKeyClass(Text.class);
        wordCountJob.setMapOutputValueClass(IntWritable.class);

        //设置最终输出的kv数据类型
        wordCountJob.setOutputKeyClass(Text.class);
        wordCountJob.setOutputValueClass(IntWritable.class);

        //设置要处理的文本数据所存放的路径
        FileInputFormat.setInputPaths(wordCountJob, args[0]);
        FileOutputFormat.setOutputPath(wordCountJob, new Path(args[1]));
        FSInputStream fis;
        //提交job给hadoop集群
        wordCountJob.submit();
        return 0;
    }

    public static void  main(String args[]) throws Exception {
        ToolRunner.run(new WordCount(),args);
    }
}
